前言

玩過 Agent 之後，你可能只用過 ADK Web 來互動，但如果想要把自己的 Agent 變成 API，就可以使用 FastAPI 來包裝它。
話不多說，先今天的內容～(oﾟ▽ﾟ)o

FastAPI 封裝 Agent：基礎回顧

在開始之前，回顧一下 FastAPI 的幾個特性：

• 快速上手：裝飾器 + Pydantic，就能定義路由和資料模型
• 自動生成文件：Swagger UI / ReDoc 幫你測試 API
• 非同步支援：async/await 讓 Agent 處理高併發也沒問題
• 易於擴充：CORS、Middleware、Dependency Injection 都支援

將 ADK Agent 包裝成 API 端點

假設你已經有一個 root_agent，我們可以透過 FastAPI 將它變成 /query 端點：

from fastapi import FastAPI
from google.genai import types
from google.adk.runners import Runner
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.sessions import InMemorySessionService
from pydantic import BaseModel
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv

from root_agent import root_agent

load_dotenv()
app = FastAPI()

# CORS 設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Session 與 Artifact 服務
session_service = InMemorySessionService()
artifacts_service = InMemoryArtifactService()

class QueryRequest(BaseModel):
    query: str

接下來定義處理 Agent 的 POST 路由：

@app.post("/query")
async def handle_query(request: QueryRequest):
    try:
        # 建立新會話
        session = await session_service.create_session(
            state={},
            app_name="my_app",
            user_id="user"
        )
        
        # 將使用者訊息轉成 ADK Content
        parts = [types.Part(text=request.query)]
        content = types.Content(role="user", parts=parts)
        
        # Runner 負責執行 Agent
        runner = Runner(
            app_name="my_app",
            agent=root_agent,
            session_service=session_service,
            artifact_service=artifacts_service
        )
        
        # 非同步執行
        events_async = runner.run_async(
            session_id=session.id,
            user_id="user",
            new_message=content
        )
        
        # 收集回應
        response = []
        async for event in events_async:
            if event.content:
                for part in event.content.parts:
                    if part.text:
                        response.append(part.text)

        return {"response": "\n".join(response)}
    
    except Exception as e:
        # 錯誤處理
        return {"error": str(e)}

非同步與錯誤回應機制

1. 非同步執行 (async)

   • FastAPI 端點用 async def，Runner 用 run_async
   • 可同時處理多個使用者請求，不會阻塞其他任務

2. 錯誤捕捉 (try/except)

   • 避免 Agent 或 Session 發生錯誤時整個 API 崩潰
   • 透過 {"error": str(e)} 回傳明確訊息給前端

測試你的 Agent API

更新requirements.txt

google-generativeai>=0.3.0
typing-extensions>=4.5.0
python-dotenv>=1.0.0
requests>=2.31.0
aiohttp>=3.8.0
fastapi>=0.104.0
uvicorn>=0.24.0
gradio>=4.0.0
google-adk
litellm

啟動 FastAPI :

uvicorn main:app --host 0.0.0.0 --port <port>

開啟瀏覽器：

http://127.0.0.1:<port>/docs
            or
http://localhost:<port>/docs

• Swagger UI 自動生成
• 可直接測試 /query POST，輸入文字訊息
• 看到 Agent 的回應立即回傳，很方便～(≧▽≦)/

結果畫面

成功可以看到 Response 回傳內容啦~
想要關閉時， 使用 Ctrl+C 就可以了

結尾廢話

看完這篇，你應該就能把自己的 Agent 變成一個可呼叫的 API 了！
未來可以加上 使用者認證、CORS 限制、背景任務 等等，讓它更完整、更安全。

我們下一篇見～(oﾟvﾟ)ノ